/////////////////////////////////////////////////////////////////////////////
// 2023 -
//open source under Apache License Version 2.0
/////////////////////////////////////////////////////////////////////////////
//  Modifications by Kumo Inc.
// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#include <kmhdfs/client/striped_input_stream_impl.h>
#include <kmhdfs/client/stateful_stripe_reader.h>
#include <kmhdfs/client/raw_erasure_decoder.h>
#include <kmhdfs/client/ec_chunk.h>
#include <kmhdfs/client/preconditions.h>
#include <turbo/log/logging.h>

namespace Hdfs {
namespace Internal {

class CorruptedBlocks;
StatefulStripeReader::StatefulStripeReader(StripedBlockUtil::AlignedStripe & alignedStripe,
                                           std::shared_ptr<ECPolicy> ecPolicy,
                                           std::vector<LocatedBlock> & targetBlocks,
                                           std::vector<StripeReader::BlockReaderInfo *> & readerInfos,
                                           std::shared_ptr<CorruptedBlocks> corruptedBlocks,
                                           std::shared_ptr<RawErasureDecoder> decoder,
                                           StripedInputStreamImpl * dfsStripedInputStream,
                                           std::shared_ptr<SessionConfig> conf) :
                                           StripeReader(alignedStripe, ecPolicy, targetBlocks,
                                                        readerInfos, corruptedBlocks, decoder,
                                                        dfsStripedInputStream, conf) {
}

StatefulStripeReader::~StatefulStripeReader() {
}

void StatefulStripeReader::prepareDecodeInputs() {
    std::shared_ptr<ByteBuffer> cur;
    {
        std::lock_guard<std::mutex> lk(mtx);
        cur = std::shared_ptr<ByteBuffer>(dfsStripedInputStream->getCurStripeBuf()->duplicate());
    }

    if (decodeInputs.empty()) {
        decodeInputs = std::vector<std::shared_ptr<ECChunk>>(dataBlkNum + parityBlkNum);
    }
    int bufLen = static_cast<int>(alignedStripe.getSpanInBlock());
    int bufOff = static_cast<int>(alignedStripe.getOffsetInBlock());
    for (int i = 0; i < dataBlkNum; i++) {
        cur->limit(cur->capacity());
        int pos = bufOff % cellSize + cellSize * i;
        cur->position(pos);
        cur->limit(pos + bufLen);
        decodeInputs[i] = std::shared_ptr<ECChunk>(
            new ECChunk(std::shared_ptr<ByteBuffer>(cur->slice()), 0, bufLen));
        if (alignedStripe.chunks[i] == nullptr) {
            alignedStripe.chunks[i] =
                std::shared_ptr<StripedBlockUtil::StripingChunk>(new StripedBlockUtil::StripingChunk(decodeInputs[i]->getBuffer()));
        }
    }
}

bool StatefulStripeReader::prepareParityChunk(int index) {
    Preconditions::checkState(index >= dataBlkNum
        && alignedStripe.chunks[index] == nullptr);
    const int parityIndex = index - dataBlkNum;
    std::shared_ptr<ByteBuffer> buf =
        std::shared_ptr<ByteBuffer>(dfsStripedInputStream->getParityBuffer()->duplicate());
    buf->position(cellSize * parityIndex);
    buf->limit(cellSize * parityIndex + static_cast<int>(alignedStripe.range->spanInBlock));
    decodeInputs[index] =
        std::shared_ptr<ECChunk>(
            new ECChunk(std::shared_ptr<ByteBuffer>(buf->slice()), 0,
                        static_cast<int>(alignedStripe.range -> spanInBlock)));
    alignedStripe.chunks[index] =
        std::shared_ptr<StripedBlockUtil::StripingChunk>(new StripedBlockUtil::StripingChunk(decodeInputs[index]->getBuffer()));
    return true;
}

void StatefulStripeReader::decode() {
    VKLOG(300)<<turbo::str_format( "decoding!!!");
    finalizeDecodeInputs();
    decodeAndFillBuffer(false);
}

/**
 * Default close do nothing.
 */
void StatefulStripeReader::close() {
}

}
}
